package com.c.rabbit_mq;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.map.TableMap;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import cn.hutool.log.StaticLog;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.tool.rabbit.ToolRabbitMQ;
import lombok.Data;
import ntd.tool.abnormal.ComAbnormal;
import ntd.util.modular.data.util.handle.tool.Tool212;

import java.io.IOException;
import java.util.List;

/**
 * mqtt 处理机制
 */
@Data
public abstract class ComRabbitMqHandle<T> implements Consumer {

    private String topic;
    private Class<T> clazz;

    public ComRabbitMqHandle(String topic, Class<T> clazz) {
        this.topic = topic;
        this.clazz = clazz;
    }

    @Override
    public void handleConsumeOk(String s) {}

    @Override
    public void handleCancelOk(String s) {}

    @Override
    public void handleCancel(String s) {}

    @Override
    public void handleShutdownSignal(String s, ShutdownSignalException e) {}

    @Override
    public void handleRecoverOk(String s) {}

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
        on_msg(new String(bytes, "UTF-8"));
    }

    /**
     * 消息中转函数
     * @param msg
     */
    public void on_msg(String msg) {
        try {
            if (JSONUtil.isTypeJSON(msg)) {
                handle(topic, JSONUtil.toBean(msg, clazz));
            } else {
                handle(topic, Convert.convert(clazz, msg));
            }
        } catch (Exception e) {
            StaticLog.error(e, "rabbit_mq Handle 执行异常, 传入数据[{}]", msg);
        }
    }

    /**
     * 启动监听
     */
    public void start_rabbit_mq_listen() {
        ToolRabbitMQ.on_msg(this.getTopic(), this);
    }

    /**
     * 具体处理实现
     * @param t
     */
    public abstract void handle(String topic, T t);

}
